package com.atguigu.windowFunction;

import com.atguigu.bean.AdsClickLog;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * Created by RUI on 2021/6/18 10:20
 * <p>
 * 对一段时间内（比如一天内）的用户点击行为进行约束，如果对同一个广告点击超过一定限额（比如100次），应该把该用户加入黑名单并报警，此后其点击行为不应该再统计
 */
public class High_prj_blackList {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<String> mainStream = env.readTextFile("input/AdClickLog.csv")
                .map(line -> {
                    String[] datas = line.split(",");
                    return new AdsClickLog(Long.parseLong(datas[0]),
                            Long.parseLong(datas[1]),
                            datas[2],
                            datas[3],
                            Long.parseLong(datas[4]) * 1000);
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.
                        <AdsClickLog>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<AdsClickLog>() {
                            @Override
                            public long extractTimestamp(AdsClickLog element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }))
                .keyBy(value -> value.getUserId() + "-" + value.getAdsId())
                .process(new KeyedProcessFunction<String, AdsClickLog, String>() {

                    private ValueState<Boolean> warnState;
                    private ValueState<String> dateState;
                    private ReducingState<Integer> reduceState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        reduceState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Integer>(
                                "reduceState",
                                Integer::sum,
                                Integer.class
                        ));
                        dateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("dateState", String.class));
                        warnState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("warnState", Boolean.class));
                    }

                    @Override
                    public void processElement(AdsClickLog element,
                                               Context ctx,
                                               Collector<String> out) throws Exception {

                        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
                        String toDay = format.format(new Date(element.getTimestamp()));

                        if (dateState.value() == null || !toDay.equals(dateState.value())){
                            reduceState.clear();
                            warnState.clear();
                            dateState.update(toDay);
                        }

                        Integer count = reduceState.get();

                        if (count == null || count < 100) {
                            reduceState.add(1);
                            count = reduceState.get();
                            String msg = ctx.getCurrentKey() + "点击量是: " + count;
                            out.collect(msg);
                        } else if (warnState.value() == null) {
                            // 把这个用户添加到黑名单, 不再统计这个用户的数据
                            String msg = ctx.getCurrentKey() + "点击量是: " + count + " 超出阈值, 加入黑名单";
                            ctx.output(new OutputTag<String>("black"){}, msg);
                            warnState.update(true);
                        }

                    }
                });

        mainStream.print("正常");
        mainStream.getSideOutput(new OutputTag<String>("black"){}).print("黑名单");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
